package com.gxljc.bear.crawler.image;

import com.gxljc.commons.util.StringUtil;
import com.gxljc.bear.crawler.base.DolphinCrawlerConsts;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.log4j.Logger;

import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

/**
 * 封装从hbase读取数据库的dao
 *
 * @author tanghaitao
 * @since 2015-10-2 下午4:31:54
 */
public class ImageHbaseDao extends Configuration implements Serializable {
    private static final long serialVersionUID = -8639430827286907828L;
    private static final Logger LOG = Logger.getLogger(ImageHbaseDao.class);
    private static HTable imageHBTL;
    private static final ImageHbaseDao dao = new ImageHbaseDao();

    private static void initConfiguration(Configuration conf) {
        conf.addResource(DolphinCrawlerConsts.FILE_HBASE_CONFIG);
        conf.addResource(DolphinCrawlerConsts.FILE_CORE_SITE);
        conf.addResource(DolphinCrawlerConsts.FILE_HDFS_SITE);
        conf.addResource(DolphinCrawlerConsts.FILE_YARN_SITE);
    }

    private ImageHbaseDao() {
        initConfiguration(this);
        try {
            imageHBTL = new HTable(this, ImageConsts.HBASE_TABLE_IMAGE);
        } catch (Exception e) {
            LOG.error("##init hbase table error!", e);
        }
    }

    public static ImageHbaseDao getInstance() {
        return dao;
    }

    public boolean saveImage2Hbase(Put imagePut) {
        boolean isSuccess = false;
        try {
            imageHBTL.put(imagePut);
            isSuccess = true;
        } catch (Exception e) {
            LOG.error("##save image into hbase error!", e);
        }
        return isSuccess;
    }

    /**
     * 根据图片url从hbase中获取图片流并返回;
     *
     * @param imageUrl
     * @return
     * @throws IOException
     * @date 2022年11月27日 下午7:33:43
     */
    public boolean checkExistImage(String imageUrl) {
        String rowKey = StringUtil.genMD5Val(imageUrl);
        Get get = new Get(Bytes.toBytes(rowKey));
        try {
            Result rs = imageHBTL.get(get);
            if (rs == null) {
                return false;
            }
            return getValueAsStream(rs, ImageConsts.HBASE_FAMILY_IMAGE_INFO,
                    ImageConsts.HBASE_QUALIFIER_IMAGE_BINARY) != null;
        } catch (Exception e) {
            LOG.error("##get image from hbase error!", e);
            return false;
        }
    }

    public ImageEntity getImageFromHbaseNotContent(String imageUrl) {
        String rowKey = StringUtil.genMD5Val(imageUrl);
        try {
            Get get = new Get(Bytes.toBytes(rowKey));
            Result result = imageHBTL.get(get);
            if (result == null || result.isEmpty()) {
                return null;
            }
            ImageEntity imageEntity = new ImageEntity();
            String width = getImageValue(result, ImageConsts.HBASE_QUALIFIER_IMAGE_WIDTH);
            if (StringUtils.isNotBlank(width)) {
                imageEntity.setWidth(Integer.parseInt(width));
            }
            String height = getImageValue(result, ImageConsts.HBASE_QUALIFIER_IMAGE_HEIGHT);
            if (StringUtils.isNotBlank(height)) {
                imageEntity.setHeight(Integer.parseInt(height));
            }
            imageEntity.setImageId(rowKey);
            imageEntity.setOriginalUrl(imageUrl);
            return imageEntity;
        } catch (Exception e) {
            LOG.error("##get image from hbase error!", e);
        }
        return null;
    }

    public ImageEntity getImageFromHbase(String imageUrl) {
        String rowKey = StringUtil.genMD5Val(imageUrl);
        try {
            Get get = new Get(Bytes.toBytes(rowKey));
            Result result = imageHBTL.get(get);
            byte[] imageBinary = result.getValue(ImageConsts.HBASE_FAMILY_IMAGE_INFO.getBytes(),
                    ImageConsts.HBASE_QUALIFIER_IMAGE_BINARY.getBytes());
            if (imageBinary == null || imageBinary.length == 0) {
                return null;
            }
            ImageEntity imageEntity = new ImageEntity();
            String width = getImageValue(result, ImageConsts.HBASE_QUALIFIER_IMAGE_WIDTH);
            if (StringUtils.isNotBlank(width)) {
                imageEntity.setWidth(Integer.parseInt(width));
            }
            String height = getImageValue(result, ImageConsts.HBASE_QUALIFIER_IMAGE_HEIGHT);
            if (StringUtils.isNotBlank(height)) {
                imageEntity.setHeight(Integer.parseInt(height));
            }
            imageEntity.setBinaryImage(imageBinary);
            imageEntity.setImageId(rowKey);
            imageEntity.setOriginalUrl(imageUrl);
            return imageEntity;
        } catch (Exception e) {
            LOG.error("##get image from hbase error!", e);
        }
        return null;
    }

    //抽取图片URL的ID
    private String extractUrlId(String url) {
        if (StringUtils.isEmpty(url)) return null;
        final String patternUrl = "getImage.id=(.*?)";
        Pattern p = Pattern.compile(patternUrl);
        Matcher m = p.matcher(url);
        if (m.find()) {
            return m.group(1);
        }
        return null;
    }

    public ImageEntity getImageFromHbaseBySize(String imageUrl) {
        String rowKey = extractUrlId(imageUrl);
        try {
            if (StringUtils.isEmpty(rowKey)) return null;
            Get get = new Get(Bytes.toBytes(rowKey));
            Result result = imageHBTL.get(get);
            ImageEntity imageEntity = new ImageEntity();
            String width = getImageValue(result, ImageConsts.HBASE_QUALIFIER_IMAGE_WIDTH);
            if (StringUtils.isNotBlank(width)) {
                imageEntity.setWidth(Integer.parseInt(width));
            }
            String height = getImageValue(result, ImageConsts.HBASE_QUALIFIER_IMAGE_HEIGHT);
            if (StringUtils.isNotBlank(height)) {
                imageEntity.setHeight(Integer.parseInt(height));
            }
            imageEntity.setImageId(rowKey);
            imageEntity.setOriginalUrl(imageUrl);
            return imageEntity;
        } catch (Exception e) {
            LOG.error("##get image from hbase error!", e);
        }
        return null;
    }

    private InputStream getValueAsStream(Result result, String family, String qualifier) {
        byte[] value = result.getValue(family.getBytes(), qualifier.getBytes());
        return null == value ? null : new ByteArrayInputStream(value);
    }

    private String getValue(Result result, String family, String qualifier) {
        byte[] value = result.getValue(family.getBytes(), qualifier.getBytes());
        if (value == null) {
            return null;
        }
        return Bytes.toString(value);
    }

    private String getImageValue(Result result, String qualifier) {
        return getValue(result, ImageConsts.HBASE_FAMILY_IMAGE_INFO, qualifier);
    }

    private List<Get> getGets(List<String> clusterIdList) {
        List<Get> getList = new ArrayList<Get>();
        for (String id : clusterIdList) {
            Get get = new Get(Bytes.toBytes(id));
            getList.add(get);
        }
        return getList;
    }
}
